use anyhow::Result;
use dashmap::DashMap;
use prost::Message;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use tokio::time::{timeout, Duration};
use tracing::{error, info, trace, warn};

use crate::constants::TIMEOUT_DURATION;
use crate::error::War3Error;
use crate::ip_pool::IpPool;
use crate::proto::{DataForward, IpAssignment, MessageType, War3Message};
use crate::utils::{receive_protobuf_message, send_protobuf_message, transfer_data};

use super::client_session::{ClientSession, SendMessage};

/// 服务器状态
pub struct ServerState {
    pub ip_pool: IpPool,
    pub clients: DashMap<String, Arc<ClientSession>>,
}

impl ServerState {
    pub fn new() -> Self {
        Self {
            ip_pool: IpPool::new(),
            clients: DashMap::new(),
        }
    }
}

/// 处理客户端连接
pub async fn handle_client(
    state: Arc<ServerState>,
    mut stream: TcpStream,
    addr: SocketAddr,
) -> Result<()> {
    // 读取消息
    let message: War3Message = receive_protobuf_message(&mut stream).await?;

    match MessageType::try_from(message.r#type) {
        Ok(MessageType::ClientConnect) => {
            handle_client_connect(state, stream, addr).await?;
        }
        Ok(MessageType::War3Connect) => {
            handle_war3_connect(state, stream, message).await?;
        }
        Ok(MessageType::ProvideWar3) => {
            handle_provide_war3(state, stream, message).await?;
        }
        _ => {
            warn!("未知消息类型: {}", message.r#type);
        }
    }

    Ok(())
}

/// 处理客户端连接请求
pub async fn handle_client_connect(
    state: Arc<ServerState>,
    mut stream: TcpStream,
    addr: SocketAddr,
) -> Result<()> {
    // 分配IP（现在是异步的）
    let war3_ip = state.ip_pool.acquire_ip().await?;
    info!("Client connected from {}, assigned War3 IP: {}", addr, war3_ip);

    // 获取其他客户端IP列表
    let other_ips: Vec<String> = state
        .clients
        .iter()
        .map(|entry| entry.key().clone())
        .collect();

    // 发送IP分配结果
    let ip_assignment = IpAssignment {
        assigned_ip: war3_ip.clone(),
        other_ips: other_ips.clone(),
    };
    let response = War3Message {
        r#type: MessageType::ClientConnectResult as i32,
        source_ip: String::new(),
        target_ip: String::new(),
        payload: ip_assignment.encode_to_vec(),
    };
    
    // 如果发送失败，需要释放IP（参考 Go 项目实现）
    if let Err(e) = send_protobuf_message(&mut stream, &response).await {
        error!("发送IP分配消息失败: {}", e);
        state.ip_pool.release_ip(war3_ip)?;
        return Err(e.into());
    }

    // 创建客户端会话
    let (session, send_rx) = ClientSession::new(war3_ip.clone(), stream);
    let session = Arc::new(session);
    state.clients.insert(war3_ip.clone(), Arc::clone(&session));

    {
        let all_ips: Vec<String> = state
            .clients
            .iter()
            .map(|entry| entry.key().clone())
            .collect();
        info!("Current connected clients: {:?}", all_ips);
    }

    // 通知所有客户端更新列表
    for client in state.clients.iter() {
        let mut client_list = state
            .clients
            .iter()
            .map(|entry| entry.key().clone())
            .collect::<Vec<_>>();

        // 为每个客户端生成排除自己的列表
        client_list.retain(|ip| ip != client.key());

        if !client_list.is_empty() {
            info!(
                "Syncing client list update to War3 IP={}, other_clients={:?}",
                client.key(),
                client_list
            );
            client.value().update_client_list(client_list).await?;
        }
    }

    {
        // 处理客户端消息
        let state_clone = Arc::clone(&state);
        let session_clone = Arc::clone(&session);
        let war3_ip_for_cleanup = war3_ip.clone();
        tokio::spawn(async move {
            if let Err(e) =
                handle_client_messages(state_clone, session_clone, send_rx).await
            {
                error!("处理客户端消息错误 {}: {}", war3_ip_for_cleanup, e);
            }

            // 清理资源
            state.clients.remove(&war3_ip_for_cleanup);
            let _ = state.ip_pool.release_ip(war3_ip_for_cleanup.clone());
            info!("客户端断开: {}", war3_ip_for_cleanup);
        });
    }

    Ok(())
}

/// 处理客户端消息循环
pub async fn handle_client_messages(
    state: Arc<ServerState>,
    session: Arc<ClientSession>,
    mut send_rx: mpsc::Receiver<SendMessage>,
) -> Result<()> {
    // 获取已经split的读写半部分（在ClientSession创建时已经split）
    // 这样读写操作不会互相竞争锁，彻底消除锁竞争问题
    let read_half = session.read_half();
    let write_half = session.write_half();

    // 创建一个任务专门处理接收消息
    let (receive_tx, mut receive_rx) = mpsc::channel::<Result<War3Message, War3Error>>(100);
    let read_half_for_receiving = Arc::clone(&read_half);
    let receive_task = tokio::spawn(async move {
        info!("Receive task started for client (using split read half - no lock contention!)");
        loop {
            // 获取读半部分的锁（不会与写半部分竞争）
            let mut reader = read_half_for_receiving.lock().await;
            info!("Receive task: acquired read lock, reading message");
            
            // 读取消息，设置合理的超时
            let receive_result: Result<Result<War3Message, War3Error>, tokio::time::error::Elapsed> = tokio::time::timeout(
                Duration::from_secs(30), // 设置30秒超时
                receive_protobuf_message(&mut *reader),
            ).await;

            // 立即释放读锁
            drop(reader);
            info!("Receive task: read lock released");

            match receive_result {
                Ok(Ok(msg)) => {
                    let msg_type = MessageType::try_from(msg.r#type)
                        .map(|t| format!("{:?}", t))
                        .unwrap_or_else(|_| "Unknown".to_string());

                    // 只对非DataForward消息记录info级别日志
                    if msg.r#type != MessageType::DataForward as i32 {
                        info!(
                            "Receive task: received message type: {}, source: {}, target: {}, payload size: {}",
                            msg_type, msg.source_ip, msg.target_ip, msg.payload.len()
                        );
                    } else {
                        info!(
                            "Receive task: received DataForward: {} -> {}, {} bytes",
                            msg.source_ip, msg.target_ip, msg.payload.len()
                        );
                    }

                    if receive_tx.send(Ok(msg)).await.is_err() {
                        warn!("Receive task: failed to send message to main loop");
                        break; // 通道已关闭
                    }
                }
                Ok(Err(e)) => {
                    error!("Receive task: error receiving message: {:?}", e);
                    let _ = receive_tx.send(Err(e)).await;
                    break;
                }
                Err(_) => {
                    info!("Receive task: timeout receiving message (30s)");
                    // 超时继续循环，不退出
                    continue;
                }
            }
        }
        info!("Receive task ended for client");
    });

    // 创建一个任务专门处理发送消息
    let write_half_for_sending = Arc::clone(&write_half);
    let send_task = tokio::spawn(async move {
        info!("Send task started for client (using split write half - no lock contention!)");
        while let Some(send_msg) = send_rx.recv().await {
            let msg_type = MessageType::try_from(send_msg.message.r#type)
                .map(|t| format!("{:?}", t))
                .unwrap_or_else(|_| "Unknown".to_string());

            // 只对非DataForward消息记录trace级别日志
            if send_msg.message.r#type != MessageType::DataForward as i32 {
                info!("Send task: processing message type: {}", msg_type);
            } else {
                info!("Send task: processing DataForward: {} -> {}", send_msg.message.source_ip, send_msg.message.target_ip);
            }
            
            info!("Send task: acquiring write lock for message type: {}", msg_type);

            // 获取写半部分的锁（不会与读半部分竞争）
            let mut writer = write_half_for_sending.lock().await;
            info!("Send task: write lock acquired for message type: {}", msg_type);

            // 使用超时来避免发送操作阻塞
            let send_result = tokio::time::timeout(
                Duration::from_secs(5), // 发送超时时间5秒
                send_protobuf_message(&mut *writer, &send_msg.message),
            )
            .await;

            let result = match send_result {
                Ok(Ok(())) => {
                    // 只对非DataForward消息记录trace级别日志
                    if send_msg.message.r#type != MessageType::DataForward as i32 {
                        info!(
                            "Send task: send_protobuf_message completed successfully for type: {}",
                            msg_type
                        );
                    } else {
                        info!(
                            "Send task: DataForward sent successfully: {} -> {}",
                            send_msg.message.source_ip, send_msg.message.target_ip
                        );
                    }
                    Ok(())
                }
                Ok(Err(e)) => {
                    error!(
                        "Send task: send_protobuf_message failed for type: {} - {:?}",
                        msg_type, e
                    );
                    Err(War3Error::ConnectionClosed)
                }
                Err(_) => {
                    error!(
                        "Send task: send_protobuf_message timeout for type: {}",
                        msg_type
                    );
                    Err(War3Error::ConnectionClosed)
                }
            };

            drop(writer); // 立即释放写锁
            info!("Send task: write lock released for message type: {}", msg_type);

            // 检查是否有错误（在移动result之前）
            let has_error = result.is_err();
            
            // 发送结果到响应通道（如果需要）
            if let Some(response_tx) = send_msg.response_tx {
                if response_tx.send(result).is_err() {
                    warn!(
                        "Send task: failed to send response for message type: {}",
                        msg_type
                    );
                } else {
                    info!(
                        "Send task: response sent successfully for message type: {}",
                        msg_type
                    );
                }
            }
            
            // 如果发送失败，退出任务
            if has_error {
                error!("Send task: exiting due to send error");
                break;
            }
        }
        info!("Send task ended for client");
    });

    // 主循环处理接收到的消息
    loop {
        info!("Main loop: waiting for message from receive task");
        match receive_rx.recv().await {
            Some(Ok(message)) => {
                let msg_type = MessageType::try_from(message.r#type)
                    .map(|t| format!("{:?}", t))
                    .unwrap_or_else(|_| "Unknown".to_string());
                info!("Main message loop: received message type: {}", msg_type);
                
                // 添加更多调试信息
                info!("Message details - type: {}, source: {}, target: {}, payload size: {}",
                      msg_type, message.source_ip, message.target_ip, message.payload.len());
                
                // 处理消息时不再持有任何锁 (Rust 2024: 使用 if-let 链式匹配)
                if let Ok(MessageType::DataForward) = MessageType::try_from(message.r#type) {
                    info!("Processing DataForward message");
                    match DataForward::decode(&message.payload[..]) {
                        Ok(data_forward) => {
                            let from_ip = data_forward.from_ip.clone(); // 发送方的虚拟IP
                            let target_ip = message.target_ip.clone(); // 目标客户端的虚拟IP
                            info!(
                                "Server received DataForward: from={} (sender), target={} (target client), data size={}",
                                from_ip, target_ip, data_forward.data.len()
                            );

                            // 查找目标客户端并转发数据
                            if let Some(target_client) = state.clients.get(&from_ip) {
                                info!("Server forwarding to target client,target_ip={},from_ip={}", target_ip, from_ip);
                                if let Err(e) = target_client
                                    .forward_data(from_ip.clone(), target_ip.clone(), data_forward.data)
                                    .await
                                {
                                    error!("Failed to forward data to client {}: {:?}", target_ip, e);
                                }
                            } else {
                                info!("Target client not found: {}", target_ip);
                            }
                        }
                        Err(e) => {
                            error!("Failed to decode DataForward message: {:?}", e);
                        }
                    }
                }
            }
            Some(Err(e)) => {
                // 接收错误，退出循环
                error!("Main loop: receive error: {:?}", e);
                break;
            }
            None => {
                // 通道关闭，退出循环
                info!("Main loop: receive channel closed");
                break;
            }
        }
    }

    // 停止任务
    receive_task.abort();
    send_task.abort();
    Ok(())
}

/// 处理War3连接请求
pub async fn handle_war3_connect(
    state: Arc<ServerState>,
    stream: TcpStream,
    message: War3Message,
) -> Result<()> {
    let target_ip = message.target_ip.clone();
    info!("War3 connection request to {}", target_ip);
    
    // 查找目标客户端 (Rust 2024: 使用 let-else 简化错误处理)
    let Some(target_client) = state.clients.get(&target_ip) else {
        return Err(War3Error::ClientNotFound(target_ip.clone()).into());
    };

    info!("Found target client, sending notification");
    // 通知目标客户端
    target_client.send_notification().await?;
    
    info!("Waiting for target client to provide connection");
    // 等待目标客户端提供连接 - 这里实际上会收到一个TCP连接
    // 需要建立双向数据转发
    let result = timeout(TIMEOUT_DURATION, async {
        let channel_queue_rx = target_client.channel_queue_rx();
        let mut rx = channel_queue_rx.lock().await;
        match rx.recv().await {
            Some(dst_stream) => {
                info!("Received target client connection, starting data transfer");
                // 建立双向数据转发
                let src_addr = stream.peer_addr().ok();
                let dst_addr = dst_stream.peer_addr().ok();
                info!("Starting bidirectional data transfer: {:?} <-> {:?}", src_addr, dst_addr);
                
                // 使用transfer_data进行双向转发
                match transfer_data(stream, dst_stream).await {
                    Ok(_) => {
                        info!("Data transfer completed: {:?} <-> {:?}", src_addr, dst_addr);
                        Ok(())
                    }
                    Err(e) => {
                        error!("Data transfer failed: {:?} <-> {:?}, error: {}", src_addr, dst_addr, e);
                        Err(e)
                    }
                }
            }
            None => {
                error!("Target client channel closed");
                Err(anyhow::anyhow!("目标客户端通道关闭").into())
            }
        }
    }).await;

    match result {
        Ok(Ok(_)) => {
            info!("War3 connection and data transfer successful: -> {}", target_ip);
            Ok(())
        }
        Ok(Err(e)) => {
            error!("War3 connection or data transfer failed: -> {}, error: {}", target_ip, e);
            Err(anyhow::anyhow!("{}", e))
        }
        Err(_) => {
            error!("War3 connection timeout: -> {}", target_ip);
            Err(War3Error::Timeout.into())
        }
    }
}

/// 处理提供War3连接
pub async fn handle_provide_war3(
    state: Arc<ServerState>,
    stream: TcpStream,
    message: War3Message,
) -> Result<()> {
    let source_ip = message.source_ip.clone();
    info!("handle_provide_war3: source_ip={}", source_ip);
    
    // 查找源客户端 (Rust 2024: 使用 let-else 简化错误处理)
    let Some(source_client) = state.clients.get(&source_ip) else {
        return Err(War3Error::ClientNotFound(source_ip.clone()).into());
    };

    info!("handle_provide_war3: 找到源客户端，放入队列");
    // 将连接放入队列
    source_client.channel_queue_tx().send(stream).await?;
    info!("War3连接提供成功: {}", source_ip);
    Ok(())
}
